package spark.es

import java.time.LocalDateTime

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark.rdd.EsSpark
import org.elasticsearch.spark.sql.EsSparkSQL

/**
  * Created by zhangbn on 2018/11/2. 
  */
object ESDataHandle {
  def main(args: Array[String]): Unit = {

//    System.setProperty("hadoop.home.dir", "E:\\hadoop-common-2.7.3-bin-master")
//    System.setProperty("HADOOP_USER_NAME", "admin")

    val startTime: Long = System.currentTimeMillis
    println(s"[${LocalDateTime.now()}]  ESDataHandle作业启动.......................  now_ms=${startTime}")

    val sparkConf = new SparkConf().setAppName("ESDataHandle")
//      .setMaster("local[*]")
    sparkConf.set("spark.default.parallelism", "100")
    sparkConf.set("cluster.name", "infinova-ES")
    sparkConf.set("es.index.auto.create", "true")
    sparkConf.set("es.nodes", "10.82.27.22")
    sparkConf.set("es.port", "9201")
    sparkConf.set("es.index.read.missing.as.empty", "true")
    sparkConf.set("es.nodes.discovery", "true")
    //    sparkConf.set("es.net.http.auth.user", "elastic") //访问es的用户名
    //    sparkConf.set("es.net.http.auth.pass", "changeme") //访问es的密码
    //    sparkConf.set("es.nodes.wan.only", "true")
    val sc = new SparkContext(sparkConf)
    //write2Es(sc) //写
    //read4Es(sc); //读

    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()

    val reader = spark.read.format("jdbc")
    reader.option("url", "jdbc:mysql://10.82.27.21:3316/big_data?characterEncoding=utf-8&useSSL=false") //数据库路径
    reader.option("driver", "com.mysql.jdbc.Driver")
    reader.option("user", "root")
    reader.option("password", "vms123")
    val sql =
      s"""
        |(select
        |	person_id,
        |	capture_time,
        |	write_time,
        |	capture_resource_id,
        |	major_capture_image_url,
        |	minor_capture_image_url,
        |	sex,
        |	age,
        |	orientation,
        |	glasses,
        |	knapsack,
        |	bag,
        |	messenger_bag,
        |	shoulder_bag,
        |	umbrella,
        |	hair,
        |	hat,
        |	mask,
        |	upper_color,
        |	upper_type,
        |	upper_texture,
        |	bottom_color,
        |	bottom_type,
        |	trolley_case,
        |	barrow,
        |	baby,
        |	feature_type,
        |	feature_code
        |from
        |	pedestrian_sm
        |limit ${args(0)})
        |AS pedestrian_sm
      """.stripMargin

    reader.option("dbtable", sql) //数据表名 或者 sql 语句
    val load: DataFrame = reader.load.repartition(Integer.valueOf(args(1)))

    EsSparkSQL.saveToEs(load, "pedestrian_sm/pedestrian_sm")

//    for(a <- 1 to 13){
//      EsSparkSQL.saveToEs(load, "pedestrian_sm/pedestrian_sm")
//    }

    val endTime: Long = System.currentTimeMillis
    println(s"[${LocalDateTime.now()}]  ESDataHandle作业结束.......................  now_ms=${endTime}")

  }


  def write2Es(sc: SparkContext) = {
    val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
    val airports = Map("OTP" -> "Otopeni", "SFO" -> "San Fran")
    var rdd = sc.makeRDD(Seq(numbers, airports))
    EsSpark.saveToEs(rdd, "book/no")
    println("--------------------End-----------------")
  }

  def read4Es(sc: SparkContext) {
    val rdd = EsSpark.esRDD(sc, "forum/article", "?q=this*")

    println("------------------rdd.count():" + rdd.count())
    rdd.foreach(line => {
      val key = line._1
      val value = line._2
      println("------------------key:" + key)
      for (tmp <- value) {
        val key1 = tmp._1
        val value1 = tmp._2
        println("------------------key1:" + key1)
        println("------------------value1:" + value1)
      }
    })
  }

}
